
package com.trydofor.id.a9cq;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import javax.annotation.concurrent.ThreadSafe;

import com.trydofor.id.core.A9CQ;
import com.trydofor.id.core.ApiExceptions;

/**
 * an atomic extendible long a9cq.
 * protected by ReadWriteLock.
 * 
 * <pre>
 * <b>write</b> action are:
 *  - next()
 *  - next(long, TimeUnit)
 *  - reset()
 *  - append(long, long, int)
 * <b>read</b> action are:
 *  - current()
 *  - remains()
 *  - toString()
 *  
 * when calling next(...), the CQ will remove the used up range 
 * if the count of range is greater than the capacity.
 * so, the capacity(N) is
 * - positive - kept the last N ranges in the end. 
 * - zero     - kept nothing
 * - negative - kept all ranges.
 * </pre>
 * 
 * @author Shi Rongjiu (www.trydofor.com)
 * @since 0.9
 */
@ThreadSafe
public final class ExtedLongA9cq implements A9CQ<Long> {
    
    // lock
    private final ReadWriteLock             rwLock       = new ReentrantReadWriteLock();
    private final Lock                      rLock        = rwLock.readLock();
    private final Lock                      wLock        = rwLock.writeLock();
    
    //
    private final boolean                   isCyclic;
    private final int                       capacity;
    
    // sync
    private final LinkedList<FixedLongA9cq> readWriteBox = new LinkedList<FixedLongA9cq>();
    private volatile int                    currentIndex = 0;
    private volatile Long                   currentValue = null;
    
    private volatile long                   entireAmount = 0;
    private volatile long                   unusedAmount = 0;
    
    /**
     * an atomic extendible long a9cq. by capacity(0).
     * 
     * @param isCyclic cyclic generation or not.
     * @see #ExtedLongA9cq(boolean, int)
     */
    public ExtedLongA9cq(boolean isCyclic) {
        this(isCyclic, 0);
    }
    
    /**
     * an atomic extendible long a9cq.
     * 
     * @param isCyclic cyclic generation or not.
     * @param capacity how many ranges can be kept in the end.
     * @throws IllegalArgumentException if got bad increment or direction.
     */
    public ExtedLongA9cq(boolean isCyclic, int capacity) {
        this.isCyclic = isCyclic;
        this.capacity = capacity;
    }
    
    /**
     * whether cyclically generate sequence.
     * 
     * @return true if cyclic.
     */
    public boolean isCyclic() {
        return isCyclic;
    }
    
    /**
     * how many ranges can be kept in the end.
     * 
     * @return the capacity
     */
    public int getCapacity() {
        return capacity;
    }
    
    /**
     * {@inheritDoc}
     */
    public boolean isAtomic() {
        return true;
    }
    
    /**
     * protected by read lock.
     * 
     * {@inheritDoc}
     */
    public long remains() {
        rLock.lock();
        try {
            return isCyclic ? entireAmount : unusedAmount;
        }
        finally {
            rLock.unlock();
        }
    }
    
    /**
     * protected by read lock.
     * 
     * {@inheritDoc}
     */
    public Long current() {
        rLock.lock();
        try {
            return currentValue;
        }
        finally {
            rLock.unlock();
        }
    }
    
    /**
     * protected by write lock.
     * 
     * {@inheritDoc}
     */
    public Long next() {
        Long r = null;
        try {
            r = next(0, null);
        }
        catch (InterruptedException e) {
            ApiExceptions.throwFakeException(e);
        }
        catch (TimeoutException e) {
            ApiExceptions.throwFakeException(e);
        }
        return r;
    }
    
    /**
     * protected by write lock.
     * 
     * {@inheritDoc}
     */
    public Long next(long timeout, TimeUnit unit) throws InterruptedException, TimeoutException {
        
        final long startedNano = System.nanoTime();
        final Thread curThread = Thread.currentThread();
        final boolean needTime = (timeout > 0 && unit != null);
        
        wLock.lock();
        final Long result;
        try {
            // pre-check
            if (needTime) {
                ApiExceptions.checkThreadInterrupted(curThread);
                ApiExceptions.checkTimeoutNano(startedNano, timeout, unit);
            }
            
            if (readWriteBox.size() <= 0) { // empty
                result = null;
                currentIndex = 0;
                unusedAmount = 0;
                entireAmount = 0;
                currentValue = null;
            }
            else {
                int varIndex = currentIndex;
                long varUnused = unusedAmount;
                long varEntire = entireAmount;
                Long varValue = null;
                
                try {
                    FixedLongA9cq cq = readWriteBox.get(varIndex);
                    varValue = cq.next(timeout, unit); // may exception
                    
                    // check result
                    if (varValue == null) { // used up
                        varIndex++;
                        if (varIndex < readWriteBox.size()) {
                            varValue = readWriteBox.get(varIndex).next(timeout, unit); // may exception
                        }
                        else {
                            if (isCyclic && readWriteBox.size() > 0) {
                                long count = 0;
                                for (FixedLongA9cq a9cq : readWriteBox) { // change state
                                    a9cq.reset();
                                    count += a9cq.remains();
                                }
                                varIndex = 0;
                                varUnused = count - 1;
                                varEntire = count;
                                varValue = readWriteBox.get(0).next(timeout, unit); // may exception
                            }
                            else {
                                varIndex = 0;
                                varUnused = 0;
                                varEntire = 0;
                                varValue = null;
                            }
                        }
                    }
                    
                    // check capacity
                    if (capacity >= 0 && capacity < readWriteBox.size() && cq.remains() == 0) {
                        readWriteBox.removeFirst(); // change state
                        cq.reset();
                        varEntire -= cq.remains();
                    }
                    
                    varUnused--;
                    result = varValue;
                }
                finally {
                    // change shared value
                    currentIndex = varIndex;
                    unusedAmount = varUnused;
                    entireAmount = varEntire;
                    currentValue = varValue;
                }
            }
        }
        finally {
            wLock.unlock();
        }
        
        // post-check
        if (needTime) {
            ApiExceptions.checkThreadInterrupted(curThread);
            ApiExceptions.checkTimeoutNano(startedNano, timeout, unit);
        }
        return result;
    }
    
    /**
     * protected by write lock.
     * 
     * {@inheritDoc}
     */
    public void reset() {
        wLock.lock();
        try {
            long count = 0;
            for (FixedLongA9cq a9cq : readWriteBox) {
                a9cq.reset(); // change state
                count += a9cq.remains();
            }
            
            // change shared value
            entireAmount = count;
            unusedAmount = count;
            currentIndex = 0;
            currentValue = null;
        }
        finally {
            wLock.unlock();
        }
    }
    
    /**
     * protected by write lock.
     * 
     * put a range to the end.
     * 
     * @param startWith the first value of range.
     * @param endupWith the last value of range.
     * @param increment the increment of range.
     * @return true if appended.
     * @throws IllegalArgumentException if got bad increment or direction.
     * @see java.util.List#add(Object)
     */
    public boolean append(long startWith, long endupWith, int increment) {
        wLock.lock();
        try {
            
            long varEntire = entireAmount;
            long varUnused = unusedAmount;
            
            // append
            FixedLongA9cq cq = new FixedLongA9cq(false, startWith, endupWith, increment);
            boolean added = readWriteBox.add(cq);
            if (added) {
                long count = cq.remains();
                varEntire += count;
                varUnused += count;
            }
            
            // remove old
            if (capacity > 0 && capacity < readWriteBox.size()) {
                while (currentIndex > 0) {
                    FixedLongA9cq od = readWriteBox.removeFirst();
                    od.reset();
                    varEntire -= od.remains();
                    currentIndex--;
                }
            }
            
            // change shared value
            entireAmount = varEntire;
            unusedAmount = varUnused;
            
            return added;
        }
        finally {
            wLock.unlock();
        }
    }
    
    /**
     * protected by read lock.
     * 
     * {@inheritDoc}
     */
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("isCyclic=");
        sb.append(isCyclic);
        sb.append(",capacity=");
        sb.append(capacity);
        rLock.lock();
        try {
            sb.append(",remains=");
            sb.append((isCyclic ? entireAmount : unusedAmount));
            sb.append(",current=");
            sb.append(currentValue);
        }
        finally {
            rLock.unlock();
        }
        return sb.toString();
    }
}
